package prober_point

import (
	"context"
	"encoding/json"
	"fmt"
	"github.com/apache/rocketmq-client-go/v2/primitive"
	"github.com/panjf2000/ants/v2"
	"gobase/rocketmq"
	"gobase/signal"
	"net"
	"path/filepath"
	"runtime"
	"strings"
	"time"
)

type ProbePointMQ struct {
	RocketCfg
	CliProducer *rocketmq.MqProducer
	ProbeBody
	pool *ants.Pool
}

// NewProbe 初始化 数据探针
func NewProbe(rocketCfg RocketCfg, pro ProbeLocation) (*ProbePointMQ, error) {

	if pro.ClientIP == "" {
		pro.ClientIP = GeLocalIP()
	}

	mq := rocketmq.NewMQProducer(rocketCfg.Topic, rocketCfg.Group, rocketCfg.Address)

	//工作池大小 5000个
	pool, err := ants.NewPool(5000)
	if err != nil {
		return nil, err
	}
	signal.AddSignalFunc(pool.Release)

	return &ProbePointMQ{
		pool:        pool,
		RocketCfg:   rocketCfg,
		CliProducer: mq,
		ProbeBody: ProbeBody{
			ProbeLocation: pro,
			ProbeData: ProbeData{
				Measurement: "",
				Fields:      make([]Field, 0),
			},
		},
	}, nil
}

// ProbeAny 日志采集点
func (l *ProbePointMQ) ProbeAny(title, traceID string, fields ...Field) {
	l.SendLogToMQ(2, "Any", title, traceID, "日志采集点", fields...)
}

// ProbeInfo 日志采集点
func (l *ProbePointMQ) ProbeInfo(title, traceID string, fields ...Field) {
	l.SendLogToMQ(2, "info", title, traceID, "日志采集点", fields...)
}

// ProbeErr 日志采集点
func (l *ProbePointMQ) ProbeErr(title, traceID string, fields ...Field) {
	l.SendLogToMQ(2, "error", title, traceID, "日志采集点", fields...)
}

// ProbeWarn 日志采集点
func (l *ProbePointMQ) ProbeWarn(title, traceID string, fields ...Field) {
	l.SendLogToMQ(2, "warn", title, traceID, "日志采集点", fields...)
}

func (l *ProbePointMQ) StatusResult(status, result string, code int) {
	//TODO
}

func (l *ProbePointMQ) SendLogToMQ(skip int, level, title, traceID, action string, fields ...Field) {

	fileName, line, funcName := "???", 0, "???"

	if l == nil {
		fmt.Println("数据采集 没有被初始化")
		return
	}

	pc, fileName, line, ok := runtime.Caller(skip)
	if !ok {
		fileName, line, funcName = "???", 0, "???"
	} else {
		funcName = runtime.FuncForPC(pc).Name()      // main(*MyStruct).foo
		funcName = filepath.Ext(funcName)            // .foo
		funcName = strings.TrimPrefix(funcName, ".") // foo
	}

	err := l.pool.Submit(func() {

		l.ProbeBody.ProbeHeader = ProbeHeader{
			TraceID:  traceID,
			Title:    title,
			Trans:    "",
			Action:   action,
			Method:   fmt.Sprintf("%s:%d", fileName, line),
			DataTime: time.Now().Format(time.RFC3339),
		}
		l.ProbeBody.ProbeStatusResult = ProbeStatusResult{
			Level:      level,
			Status:     "",
			StatusCode: 0,
		}
		l.ProbeBody.ProbeData = ProbeData{
			Measurement: funcName,
			Fields:      fields,
		}

		b, err := json.Marshal(&l.ProbeBody)
		if err != nil {
			fmt.Println("json解析异常")
			return
		}

		MqMsg := rocketmq.NewMQMessage(l.Topic, l.Group, l.ServiceName, b, 0)
		_, err = l.CliProducer.SendAsyncMessage(MqMsg, func(ctx context.Context, result *primitive.SendResult, e error) {
			if e != nil {
				fmt.Println("MQ 异步发送消息 err: ", e)
				return
			}
		})

		if err != nil {
			fmt.Println("MQ 异步发送消息 err: ", err)
			return
		}
	})
	if err != nil {
		fmt.Println("ants 异步发送MQ消息 err: ", err)
	}
}

func ProbeAny(key string, val any) Field {
	return Field{
		Key:   key,
		Value: val,
	}
}

// GeLocalIP 获取机器的ip地址
func GeLocalIP() string {
	addrs, err := net.InterfaceAddrs()
	if err != nil {
		fmt.Println("get net interface address failed, err = ", err.Error())
		return ""
	}
	for _, addr := range addrs {
		if ip, ok := addr.(*net.IPNet); ok && !ip.IP.IsLoopback() {
			if ip.IP.To4() != nil {
				return ip.IP.String()
			}
		}
	}
	return ""
}
